package com.lockforzookeeper.common;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;

/**
 * zookeeper分布式锁
 * <p>
 * Watcher 观察器
 *
 * @author: 顾志杰
 */
@Slf4j
public class ZookeeperLock implements AutoCloseable, Watcher {

    private ZooKeeper zooKeeper;

    /**
     * 瞬时有序节点
     */
    private String znode;

    public ZookeeperLock() throws IOException {
        this.zooKeeper = new ZooKeeper(
                /*zookeeper地址集群用，分割*/
                "192.168.35.133:2181",
                /*超时时间 10秒*/
                100000,
                /*观察器 因为已经实现了Watcher 所以传入自己就可以了*/
                this
        );
    }

    /**
     * 获取锁
     *
     * @param businessCode 业务代码分类
     * @return
     */
    public boolean getLock(String businessCode) {
        try {
            /*创建业务根节点*/
            /*1.获取业务根节点*/
            Stat exists = zooKeeper.exists(
                    /*节点名称*/
                    "/" + businessCode,
                    /*false不需要观察器，如果需要可以自行添加*/
                    false);
            /*2.如果不存在这个节点 就创建一下业务根节点*/
            if (exists == null) {
                zooKeeper.create(
                        /*节点路径*/
                        "/" + businessCode,
                        /*节点的初始化数据*/
                        businessCode.getBytes(),
                        /*zookeeper权限  OPEN_ACL_UNSAFE不需要用户密码连接zookeeper*/
                        ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        /*创建模式  PERSISTENT持久节点*/
                        CreateMode.PERSISTENT);
            }

            /*创建瞬时有序节点*/
            znode = zooKeeper.create(
                    /*节点路径*/
                    "/" + businessCode + "/" + businessCode + "_",
                    /*节点的初始化数据*/
                    businessCode.getBytes(),
                    /*zookeeper权限  OPEN_ACL_UNSAFE不需要用户密码连接zookeeper*/
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    /*创建模式  EPHEMERAL_SEQUENTIAL瞬时有序节点*/
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            /*获取业务节点下所有子节点*/
            List<String> childrenNodes = zooKeeper.getChildren(
                    /*要获取的父节点*/
                    "/" + businessCode,
                    /*不需要观察器*/
                    false);
            /*1.排序 得到一个升序的队列*/
            Collections.sort(childrenNodes);
            /*比较节点*/
            /*1.获取第一个节点*/
            String firstNode = childrenNodes.get(0);
            /*2.如果相同后缀证明创建的节点是第一个子节点则获得锁返回true*/
            if (znode.endsWith(firstNode)) {
                return true;
            }
            /*3.不是第一个节点，则监听前一个节点*/
            String lastNode = firstNode;
            for (String childrenNode : childrenNodes) {
                if (znode.endsWith(childrenNode)) {
                    /*添加监听器*/
                    zooKeeper.exists("/" + businessCode + "/" + lastNode, true);
                    break;
                } else {
                    lastNode = firstNode;
                }
            }
            /*等待前面线程结束 在唤醒自己这个线程 等待过程需要加synchronized锁*/
            synchronized (this){
                wait();
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }


    @Override
    public void close() throws Exception {
        /*删除节点*/
        zooKeeper.delete(znode,-1);
        /*关闭*/
        zooKeeper.close();
        log.info("释放锁！！！");
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        /*如果事件节点是删除的话就唤醒线程*/
        if(watchedEvent.getType() == Event.EventType.NodeDeleted){
            synchronized (this){
                notify();
            }
        }
    }
}
